package qdu.edu.cn.storm.redis.write;

import org.apache.storm.coordination.BatchOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBatchBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import redis.clients.jedis.Jedis;

import java.util.Map;

import static qdu.edu.cn.storm.ConstantFinalString.Fields.*;
import static qdu.edu.cn.storm.ConstantFinalString.HOST;
import static qdu.edu.cn.storm.ConstantFinalString.Redis.*;

public class RedisStoreRequestListBolt extends BaseBatchBolt<Long> {
    // redis 一主二从 master：6381 slave：6379，6380
    private Jedis jedisRead;
    private Jedis jedisWrite;
    private Long userId;
    // 仅仅传""，而不传初始的超链接，这样可以解决null问题，并且数据量也很小
    private final StringBuilder hyperlinks = new StringBuilder();
    private BatchOutputCollector collector;

    @Override
    public void prepare(Map<String, Object> conf, TopologyContext context, BatchOutputCollector collector, Long id) {
        jedisRead = new Jedis(HOST, REDIS_READ_PORT);
        jedisWrite = new Jedis(HOST, REDIS_WRITE_PORT);
        this.userId = id;
        this.collector = collector;
    }

    /**
     * 根据关键词获取对应图片的超链接
     */
    @Override
    public void execute(Tuple tuple) {
        // 获取USER_ID
        String key = "list_"+userId;
        // 获取对应的关键词
        String word = tuple.getStringByField(WORD);
        // 在HASH_HYPERLINK_KEY中根据关键词（WORD）获取超链接，若不存在，就丢弃。
        String hyperlink = jedisRead.hget(HASH_HYPERLINK_KEY, word);
        if (hyperlink != null) {
            jedisWrite.lpush(key, hyperlink);
            hyperlinks.append(" ").append(hyperlink);
        }
    }

    @Override
    public void finishBatch() {
        collector.emit(new Values(userId, hyperlinks.toString()));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields(USER_ID, HYPERLINK));
    }

}
